package edu.xhu.homepage.job;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.xxl.job.core.handler.annotation.XxlJob;
import edu.xhu.api.id.IdClient;
import edu.xhu.homepage.mapper.BiliRecommendBloomMapper;
import edu.xhu.model.common.dtos.ResponseResult;
import edu.xhu.model.common.enums.AppHttpCodeEnum;
import edu.xhu.model.video.pojos.BiliRecommendBloom;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.List;
import java.util.concurrent.CompletableFuture;

@Component
@Slf4j
@RequiredArgsConstructor
public class BloomJob {

    private final StringRedisTemplate redisTemplate;

    private final BiliRecommendBloomMapper biliRecommendBloomMapper;

    private final IdClient idClient;


    /**
     * 刷游客,用户bitMap
     * 用户布隆需要持久化,游客布隆不需要持久化
     */
    @XxlJob("bloomHandler")
    @Transactional
    public List<String> scanKeys() {
        //扫描用户
        return redisTemplate.execute((RedisCallback<List<String>>) connection -> {
            try {
                //1.创建一个游标,每次扫描1000个
                ScanOptions bloomKeys = ScanOptions.scanOptions().match("bloom*").count(1000).build();
                Cursor<byte[]> scan = connection.scan(bloomKeys);
                BiliRecommendBloom bloom = new BiliRecommendBloom();
                //2.优先处理游客
                handlerBloom(999L,
                        biliRecommendBloomMapper
                                .selectOne(Wrappers.<BiliRecommendBloom>lambdaQuery()
                                        .eq(BiliRecommendBloom::getUserId, 999L))
                        , bloom, Base64.getEncoder().encodeToString(connection.get("bloom:999".getBytes())));
                //3.处理bloom
                while (scan.hasNext()) {
                    byte[] key = scan.next();
                    //处理这个用户的key,把这个key刷出来,批量存储DB
                    String bitMap = Base64.getEncoder().encodeToString(connection.get(key));
                    //异步执行
                    CompletableFuture.runAsync(() -> {
                        Long userId = Long.parseLong(new String(key, StandardCharsets.UTF_8).split(":")[1]);
                        BiliRecommendBloom one = biliRecommendBloomMapper.selectOne(Wrappers.<BiliRecommendBloom>lambdaQuery().eq(BiliRecommendBloom::getUserId, userId));
                        handlerBloom(userId, one, bloom, bitMap);
                    });
                }
                //关闭资源
                scan.close();
                connection.close();
            } catch (RuntimeException | IOException e) {
                log.error("id生成错误:{}", e.toString());
            }
            return null;
        });
    }

    private void handlerBloom(Long userId, BiliRecommendBloom one, BiliRecommendBloom bloom, String bitMap) {
        if (one == null) {
            //游客布隆,one必为null
            if (userId == 999L) {
                //刷新游客bloom
                redisTemplate.opsForValue().setBit("bloom:999", 0L, false);
                return;
            }
            //获取idClient
            ResponseResult<?> responseResult = idClient.getSnowflakeDistributedId();
            if (!responseResult.getCode().equals(AppHttpCodeEnum.SUCCESS.getCode())) {
                throw new RuntimeException();
            }
            Long id = (Long) responseResult.getData();
            bloom.setId(id).setUserId(userId).setBitmap(bitMap);
            if (biliRecommendBloomMapper.insert(bloom) < 1) {
                throw new RuntimeException();
            }
        } else {
            //不为空,执行update
            one.setBitmap(bitMap);
            if (biliRecommendBloomMapper.updateById(one) < 1) {
                throw new RuntimeException();
            }
        }
    }

}
